package com.pisolution.scheduler.runtime.master.processor;

import com.pisolution.scheduler.core.utils.Stopper;
import com.pisolution.scheduler.datasource.instance.ExecutionState;
import com.pisolution.scheduler.datasource.instance.TaskInstance;
import com.pisolution.scheduler.runtime.process.ProcessService;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 任务状态更新器 用于更新当前受追踪的任务实例的状态.
 *
 * @author sora
 * @version 0.1.0
 */
@Component
public class TaskStateUpdater implements Runnable {
    private static final Logger LOGGER = LoggerFactory.getLogger(TaskStateUpdater.class);

    // -------------------------------------------------------------------------------------------------

    /** 用于缓存待处理的任务状态转换信息. */
    private final BlockingQueue<TaskStateEvent> queue;

    /** 用于更新任务状态. */
    private final ProcessService service;

    /** 用于关闭创建的线程. */
    private Thread updateThread;

    // -------------------------------------------------------------------------------------------------
    //  初始化
    // -------------------------------------------------------------------------------------------------

    public TaskStateUpdater(final ProcessService service) {
        this.queue = new LinkedBlockingQueue<>(7000);
        this.service = service;
    }

    // -------------------------------------------------------------------------------------------------
    //  生命周期
    // -------------------------------------------------------------------------------------------------

    /**
     * 启动任务状态更新器.
     *
     * <p><b>注意:</b></p>
     * <p>该方法会在构造函数与依赖注入之后自动调用 无需手动触发.</p>
     */
    @SuppressWarnings("unused")
    @PostConstruct
    public void start() {
        this.updateThread = new Thread(this);

        this.updateThread.setName("task-state-updater");
        this.updateThread.start();
    }

    /**
     * 停止任务状态更新器 剩下未处理的事件一次性处理完.
     *
     * <p><b>注意:</b></p>
     * <p>该方法无需手动触发 销毁前会自动调用.</p>
     */
    @SuppressWarnings("unused")
    @PreDestroy
    public void stop() {
        this.updateThread.interrupt();

        if (!this.queue.isEmpty()) {
            final List<TaskStateEvent> remainingEvents = new ArrayList<>(this.queue.size());

            this.queue.drainTo(remainingEvents);
            for (final TaskStateEvent event: remainingEvents) {
                this.resolveEvent(event);
            }
        }
    }

    // -------------------------------------------------------------------------------------------------
    //  状态追加
    // -------------------------------------------------------------------------------------------------

    /**
     * 接收任务状态更改事件并等待处理.
     *
     * @param event 任务状态更改事件.
     */
    public void addEvent(final TaskStateEvent event) {
        try {
            this.queue.put(event);
        } catch (InterruptedException e) {
            LOGGER.error("Could not resolve the state change event {}.", event, e);
        }
    }

    // -------------------------------------------------------------------------------------------------
    //  状态处理
    // -------------------------------------------------------------------------------------------------

    @Override
    public void run() {
        while (Stopper.isRunning()) {
            try {
                final TaskStateEvent event = this.queue.take();

                this.resolveEvent(event);
            } catch (InterruptedException ignored) {
                break;
            } catch (Exception e) {
                LOGGER.error("Could not resolve the task state change event.", e);
            }
        }

        LOGGER.info("The task state updater is stopped.");
    }

    /**
     * 处理任务状态更新事件.
     *
     * @param event 待处理的任务状态更新事件.
     */
    private void resolveEvent(final TaskStateEvent event) {
        // 1. 确认事件是否来源于确认指令.
        if (event.isAck()) {
            try {
                final TaskInstance instance = this.service.findTaskInstanceById(event.getTaskId());

                // 2. 如果该任务实例存在 则更改其运行状态.
                if (instance != null) {
                    instance.setHost(event.getHost());
                    instance.setExecutePath(event.getExecutePath());
                    instance.setLogPath(event.getLogPath());
                    instance.setState(instance.getState().isFinished() ? instance.getState() : event.getState());
                    instance.setStartTime(event.getStartTime());

                    this.service.saveTaskInstance(instance);
                }

                // 3. 通知工作节点 任务确认指令已经处理结束.
                final DBAckCommand command = new DBAckCommand();

                command.setTaskId(event.getTaskId());
                command.setStatus(ExecutionState.SUCCESS.getCode());

                event.getChannel().writeAndFlush(command.convert());
            } catch (Exception e) {
                LOGGER.error("Could not resolve the task ack command.", e);

                final DBAckCommand command = new DBAckCommand();

                command.setTaskId(-1);
                command.setStatus(ExecutionState.FAILURE.getCode());

                event.getChannel().writeAndFlush(command.convert());
            }

            return;
        }

        // TODO: 完成此处逻辑.
    }
}
